package com.junjie.food.orderServiceManager.config;

import com.junjie.food.orderServiceManager.dto.OrderMessageDTO;
import com.junjie.food.orderServiceManager.service.OrderMessageService;
import com.junjie.food.orderServiceManager.service.impl.OrderMessageServiceImpl;
import com.junjie.food.orderServiceManager.service.impl.testMessage;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.ClassMapper;
import org.springframework.amqp.support.converter.DefaultJackson2JavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

//@Configuration
//@ConditionalOnProperty("mountainmq.enable")
@Slf4j
public class RabbitConfig {
//    @Autowired
//    private OrderMessageService orderMessageService;

    // 我草，竟然加@Autowired也可以自动调用方法
    // 就是所有加了@Configuration类中的方法如果加了@Autowired，就会自动执行该方法
//    @Autowired
//    public void startListenMessage() throws IOException, InterruptedException, TimeoutException {
//        // 因为这个方法有@Async注解，所以会异步调用，然后异步线程从我们定义的线程池里面取
//        orderMessageService.handleMessage();
//    }

    /*---------------------restaurant---------------------*/
    @Bean
    public Exchange exchange1() {
        return new DirectExchange("exchange.order.restaurant");
    }

    @Bean
    public Queue queue1() {
        return new Queue("queue.order");
    }

    @Bean
    public Binding binding1() {
        return new Binding(
                "queue.order",
                Binding.DestinationType.QUEUE,
                "exchange.order.restaurant",
                "key.order",
                null);
    }

    /*---------------------deliveryman---------------------*/
    @Bean
    public Exchange exchange2() {
        return new DirectExchange("exchange.order.deliveryman");
    }

    @Bean
    public Binding binding2() {
        return new Binding(
                "queue.order",
                Binding.DestinationType.QUEUE,
                "exchange.order.deliveryman",
                "key.order",
                null);
    }


    /*---------settlement---------*/
    @Bean
    public Exchange exchange3() {
        return new FanoutExchange("exchange.order.settlement");
    }

    @Bean
    public Exchange exchange4() {
        return new FanoutExchange("exchange.settlement.order");
    }

    @Bean
    public Binding binding3() {
        return new Binding(
                "queue.order",
                Binding.DestinationType.QUEUE,
                "exchange.settlement.order",
                "key.order",
                null);
    }

    /*--------------reward----------------*/
    @Bean
    public Exchange exchange5() {
        return new TopicExchange("exchange.order.reward");
    }

    @Bean
    public Binding binding4() {
        return new Binding(
                "queue.order",
                Binding.DestinationType.QUEUE,
                "exchange.order.reward",
                "key.order",
                null);
    }

    @Bean
    // 注意这里注入的必须是实现类，因为只有类才能继承抽象类，接口是不能继承的
    public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory,
                                                                         OrderMessageServiceImpl orderMessageService) {

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueueNames("queue.order");
        container.setExposeListenerChannel(true);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(orderMessageService);
        return container;
    }

    @Bean
    // 注意这里注入的必须是实现类，因为只有类才能继承抽象类，接口是不能继承的
    public SimpleMessageListenerContainer testMessageListener(ConnectionFactory connectionFactory,
                                                                         testMessage testMessage) {

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueueNames("test");
        container.setExposeListenerChannel(true);
//        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(testMessage);
        return container;
    }


    //====================下面的配置是可以替换成在配置文件里面去设置的=====================
//    @Bean
//    public ConnectionFactory connectionFactory(){
//        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
//        connectionFactory.setHost("110.40.176.175");
//        connectionFactory.setPort(5672);
//        connectionFactory.setPassword("guest");
//        connectionFactory.setUsername("guest");
//        // 这里用一下这个对象，AMQP才会帮你创建队列之类的
//        connectionFactory.createConnection();
//        // 开启消息确认和返回的话，下面两行代码必须要 然后消息确认返回设置成CORRELATED，否则不知道谁是谁
//        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
//        connectionFactory.setPublisherReturns(true);
//        return connectionFactory;
//    }
//
//    @Bean
//    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
//        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
//        // 加上这个值后，就会启动时候直接创建队列之类的
//        rabbitAdmin.setAutoStartup(true);
//        return rabbitAdmin;
//    }
//
//    // 注册RabbitTemplate
//    @Bean
//    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
//        // 因为需要连接，所以必须传递connectionFactory
//        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//        // 设置托管 以前设置的是channel，现在设置rabbitAdmin 设置为true后，消息若路由异常，就回调消息返回方法
//        rabbitTemplate.setMandatory(true);
//        // 设置消息返回的回调方法 就是消息无法被路由的时候
//        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
//            log.info("message:{}, replyCode:{}, replyText:{}, exchange:{}, routingKey:{}",
//                    message, replyCode, replyText, exchange, routingKey);
//        });
//        // 设置消息确认回调方法 就是消息正常发出去之后，回调该方法
//        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) ->
//                log.info("correlationData:{}, ack:{}, cause:{}",
//                        correlationData, ack, cause));
//        return rabbitTemplate;
//    }
//
//    // RabbitListener SpringBoot监听消息的终极方案
//    @Bean
//    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
//        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
//        factory.setConnectionFactory(connectionFactory);
//        return factory;
//    }
//
//    // 创建监听器 这个是比较原始的，和上面那个RabbitListener保留一个
////    @Bean
//    public SimpleMessageListenerContainer messageListenerContainer(
//            @Autowired ConnectionFactory connectionFactory
//    ) {
//        // 直接 new
//        SimpleMessageListenerContainer messageListenerContainer =
//                new SimpleMessageListenerContainer(connectionFactory);
//        // 这里可以监听多个队列，就是说可以传多个字符串
//        messageListenerContainer.setQueueNames("queue.order");
//        // 并发消费者，就是同时有几个消费者线程，在消费队列，相当于我们之前线程池的线程数 这里给3够了
//        messageListenerContainer.setConcurrentConsumers(3);
//        // 最大并发消费者线程
//        messageListenerContainer.setMaxConcurrentConsumers(5);
//
//        //========================自动确认========================
//        // 确认方式 这里我们先设置的手自动确认
////        messageListenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);
//
//        //==================快速定义收到消息的处理==================
////        messageListenerContainer.setMessageListener(new MessageListener() {
////            @Override
////            public void onMessage(Message message) {
////                // 可以改成业务操作
////                log.info("message:{}", message);
////            }
////        });
//
//        //========================手动确认========================
//        messageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
//                        messageListenerContainer.setMessageListener(new ChannelAwareMessageListener() {
//                            @Override
//                            public void onMessage(Message message, Channel channel) throws Exception {
//                                log.info("message:{}", message);
//                                // 发送消息
////                                orderMessageService.handleMessage(message.getBody());
//                                // 第二个参数是multiple，跟之前一样的
//                                channel.basicAck(
//                                        message.getMessageProperties().getDeliveryTag(),
//                                        false
//                                );
//                            }
//                        });
//        // 消费端限流 QOS
//        messageListenerContainer.setPrefetchCount(1);
//
//        // ================MessageListenerAdapter简单模式START================
//        // 使用MessageListenerAdapter，传入业务对象，自动会调用业务代码
//        // 很神奇，源码是先拿到Delegate，然后找方法里面是不是有叫 handleMessage 的方法，有就自动调用
//        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter();
//
//        // ================MessageListenerAdapter高阶模式START================
//        // 如果要塞100个，就设置128 原则是128的0.75，如果能装下100就ok，装不下就继续加
//        Map<String, String> methodMap = new HashMap<>(8);
//        // 这里就可以自定义了 键是队列 值是方法名 方法名随便自定义，只要对的上都可以自动调用
//        // 这里我们设置了ikun，只要识别到了就会自动调用
//        methodMap.put("queue.order","ikun");
//        messageListenerAdapter.setQueueOrTagToMethodName(methodMap);
//        // ================MessageListenerAdapter高阶模式END================
//
//        // Delegate是代理和代表的意思 默认会自动调用方法名为handleMessage的
////        messageListenerAdapter.setDelegate(orderMessageService);
//        // 放入container
////        messageListenerContainer.setMessageListener(messageListenerAdapter);
//        // ================MessageListenerAdapter简单模式END================
//
//
//        // 自定义消息转换 这里是自定义mq里面的消息转换，默认是传递的byte数组 当然我们也可以指定消息转换器，如下
//        // 这个转换器默认转换出来的对象是Map
////        Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter();
////        messageConverter.setClassMapper(new ClassMapper() {
////            @Override
////            public void fromClass(Class<?> clazz, MessageProperties properties) {
////            }
////            @Override // 自己决定该消息转换成什么类型
////            public Class<?> toClass(MessageProperties properties) {
////                return OrderMessageDTO.class;
////            }
////        });
////        messageListenerAdapter.setMessageConverter(messageConverter);
//
//        // 另一种方式设置消息转换器 不建议用
////        Jackson2JavaTypeMapper Jackson2JavaTypeMapper = new DefaultJackson2JavaTypeMapper();
////        messageConverter.setJavaTypeMapper(Jackson2JavaTypeMapper);
//
//        // 对象肯定是要返回回去的咯
//        return messageListenerContainer;
//    }





    // 用RabbitAdmin显示配置
//    @Autowired
//    public void initRabbit(){
//        // 点进去看源码 看继承关系 看到继承了ConnectionFactory 那跟我们之前一直new的ConnectionFactory是父子吗？
//        // 这里的ConnectionFactory是amqp包的 之前就是普通的client包下的，所以并非父子关系
//        // 但是里面其实用到了client包的ConnectionFactory，把它当做成员变量去用
//        // 然后它继承的是新的 amqp包下的 ConnectionFactory
//        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
//        connectionFactory.setHost("110.40.176.175");
//        connectionFactory.setPort(5672);
//        // 这里必须显式设置用户名和密码
//        connectionFactory.setPassword("guest");
//        connectionFactory.setUsername("guest");
//
//        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
//        /*---------------------restaurant---------------------*/
//        Exchange exchange = new DirectExchange("exchange.order.restaurant");
//        rabbitAdmin.declareExchange(exchange);
//
//        Queue queue = new Queue("queue.order");
//        rabbitAdmin.declareQueue(queue);
//
//        Binding binding = new Binding(
//                "queue.order",
//                Binding.DestinationType.QUEUE,
//                "exchange.order.restaurant",
//                "key.order",
//                null);
//
//        rabbitAdmin.declareBinding(binding);
//
//        /*---------------------deliveryman---------------------*/
//        exchange = new DirectExchange("exchange.order.deliveryman");
//        rabbitAdmin.declareExchange(exchange);
//        binding = new Binding(
//                "queue.order",
//                Binding.DestinationType.QUEUE,
//                "exchange.order.deliveryman",
//                "key.order",
//                null);
//        rabbitAdmin.declareBinding(binding);
//
//
//        /*---------settlement---------*/
//        exchange = new FanoutExchange("exchange.order.settlement");
//        rabbitAdmin.declareExchange(exchange);
//        exchange = new FanoutExchange("exchange.settlement.order");
//        rabbitAdmin.declareExchange(exchange);
//        binding = new Binding(
//                "queue.order",
//                Binding.DestinationType.QUEUE,
//                "exchange.order.settlement",
//                "key.order",
//                null);
//        rabbitAdmin.declareBinding(binding);
//
//
//        /*--------------reward----------------*/
//        exchange = new TopicExchange("exchange.order.reward");
//        rabbitAdmin.declareExchange(exchange);
//        binding = new Binding(
//                "queue.order",
//                Binding.DestinationType.QUEUE,
//                "exchange.order.reward",
//                "key.order",
//                null);
//        rabbitAdmin.declareBinding(binding);
//    }
}
